<?php
error_reporting(E_ALL);
require_once('simple-dal.php');

interface ShardKeyMapper {
        function __construct($config, $no_change_db);
        function create_store();
        function get_shards($schema_name = null);
        function add_shard($shard_name, $shard, $coord_shard = true, $accepts_new_rows = true, $extra_info = "NULL", $schema_name = null);
        function refresh_shard_cache($schema_name = null);
        function remove_shard($shard_name, $schema_name = null);
        function drop_store();
        function add_schema($schema_name, $is_default_schema = true);
        function add_column($column, $schema_name = null,$datatype = 'integer',$next_value=1);
        function move_key($column, $key, $tables, $dest_shard, $schema_name = null );
        function lock_key($column, $key, $lock_mode, $schema_name = null);
	function unlock_key($column, $key, $schema = null); 
        function get_column_info($column_name, $schema_name = null);
        function new_key($column, $key = null, $shard_name = null, $schema_name = null, $col_info = null);
        function &map($column, $key, $schema_name=null, $operator = "=", $col_info  = null);
        function &get_keys($column, $shard_name, $schema_name = null, $col_info = null);
        function &get_key_distribution($column, $schema_name = null);
        function get_column($shema_name = null);

}


class DirectoryShardKeyMapper implements ShardKeyMapper{
	var $shard_cache = array();
	var $conn = null;  

	#The constructor accepts an array which includes host, user, password and db
	public function __construct($config, $no_change_db = false) {

		if(empty($config['host']) || empty($config['user']) || empty($config['db'])) {
			throw new Exception('Missing one of required options: host, user or db');
		}

		$this->config = $config;

		$this->connect($config, $no_change_db);
	} 
	
	private function connect($config, $no_change_db = false) {
		if($this->conn) return $this->conn;
		$this->conn = SimpleDAL::factory($config);
		
		if($this->conn->my_error()) throw new Exception($this->conn->my_error());
	
		if(!$no_change_db) { 
			$this->conn->my_select_db($this->config['db']);
			if($this->conn->my_error()) throw new Exception($this->conn->my_error());
		}
	}

	private function execute($sql, $check_for_errors = true) {
		$stmt = $this->conn->my_query($sql);
	
		if($check_for_errors) {
			if(!$stmt) throw new Exception('Error while running:' . "\n$sql\n" . $this->conn->my_error());
		}

		return $stmt;
	}


	public function escape($val) {
		return $this->conn->my_escape_string($val);
	}
	
	#create the shard data store online (this is used for initial setup)
	public function create_store() {

		if(!$this->changedb($this->config['db'],true)) {
			$this->execute("CREATE DATABASE `" . $this->config['db'] . "`");
			$this->conn->my_select_db($this->config['db']);
		}

$sql = "
CREATE TABLE `schemata` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `schema_name` varchar(50) NOT NULL,
  `is_default_schema` tinyint(1) NOT NULL DEFAULT '0',
  `last_updated` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=10 DEFAULT CHARSET=latin1;
";
		echo "Creating table: schemata\n";
		$this->execute($sql);



$sql = "
CREATE TABLE `shards` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `schema_id` int(11) NOT NULL,
  `shard_name` varchar(50) NOT NULL,
  `extra_info` text,
  `shard_rdbms` enum('mysql','pgsql') NOT NULL DEFAULT 'mysql',
  `coord_shard` tinyint(1) NOT NULL DEFAULT '1',
  `accepts_new_rows` tinyint(1) NOT NULL DEFAULT '1',
  `username` varchar(64) DEFAULT NULL,
  `password` varchar(255) DEFAULT NULL,
  `host` varchar(255) DEFAULT NULL,
  `port` smallint(6) DEFAULT NULL,
  `db` varchar(45) DEFAULT NULL,
  `enabled` tinyint(1) DEFAULT '1',
  `last_updated` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `schema_id` (`schema_id`,`shard_name`),
  CONSTRAINT `shards_ibfk_1` FOREIGN KEY (`schema_id`) REFERENCES `schemata` (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=86 DEFAULT CHARSET=latin1;
";
		echo "Creating table: shards\n";
		$this->execute($sql);

$sql = "
CREATE TABLE `schemata_config` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `schema_id` int(11) NOT NULL,
  `var_name` varchar(20) NOT NULL,
  `var_value` text NOT NULL,
  `last_updated` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  KEY `schema_id` (`schema_id`,`var_name`),
  CONSTRAINT `schemata_config_ibfk_1` FOREIGN KEY (`schema_id`) REFERENCES `schemata` (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=26 DEFAULT CHARSET=latin1;
";
		echo "Creating table: schemata_config\n";
		$this->execute($sql);


$sql = "
CREATE TABLE `column_sequences` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `schema_id` int(11) NOT NULL,
  `sequence_type` enum('shard_column','sequence') NOT NULL DEFAULT 'shard_column',
  `sequence_name` varchar(50) NOT NULL,
  `next_value` bigint(20) DEFAULT NULL,
  `datatype` enum('string','integer') NOT NULL DEFAULT 'integer',
  `last_updated` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `sequence_name` (`sequence_name`,`sequence_type`),
  KEY `next_value` (`next_value`),
  KEY `schema_id` (`schema_id`),
  CONSTRAINT `column_sequences_ibfk_1` FOREIGN KEY (`schema_id`) REFERENCES `schemata` (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=5 DEFAULT CHARSET=latin1;
";

		echo "Creating table: column_sequences\n";
		$this->execute($sql);

$sql = "
CREATE TABLE `shard_map` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `column_id` int(11) NOT NULL,
  `key_value` bigint(20) DEFAULT NULL,
  `key_string_value` varchar(255) DEFAULT NULL,
  `shard_id` int(11) NOT NULL,
  `last_updated` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `column_id` (`column_id`,`key_value`,`key_string_value`),
  KEY `key_string_value` (`key_string_value`,`column_id`),
  KEY `shard_id` (`shard_id`),
  CONSTRAINT `shard_map_ibfk_1` FOREIGN KEY (`shard_id`) REFERENCES `shards` (`id`),
  CONSTRAINT `shard_map_ibfk_2` FOREIGN KEY (`column_id`) REFERENCES `column_sequences` (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=10 DEFAULT CHARSET=latin1;
";
		echo "Creating table: shard_map\n";
		$this->execute($sql);

$sql = "
CREATE TABLE `gearman_function_names` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `function_name` varchar(50) NOT NULL,
  `last_updated` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `function_name` (`function_name`)
) ENGINE=InnoDB AUTO_INCREMENT=4 DEFAULT CHARSET=latin1;
";
		echo "Creating table: gearman_function_names\n";
		$this->execute($sql);

$sql = "
CREATE TABLE `gearman_functions` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `schema_id` int(11) NOT NULL,
  `worker_count` int(10) unsigned NOT NULL,
  `is_enabled` tinyint(4) NOT NULL DEFAULT '1',
  `last_updated` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  `function_name_id` int(11) NOT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `schema_worker_unique` (`schema_id`,`function_name_id`),
  KEY `fk_gearmand_functions_schema_id_idx` (`schema_id`),
  KEY `gearman_functions_ibfk_1` (`function_name_id`),
  CONSTRAINT `gearman_functions_ibfk_1` FOREIGN KEY (`function_name_id`) REFERENCES `gearman_function_names` (`id`),
  CONSTRAINT `fk_gearmand_functions_schema_id` FOREIGN KEY (`schema_id`) REFERENCES `schemata` (`id`) ON DELETE NO ACTION ON UPDATE NO ACTION
) ENGINE=InnoDB AUTO_INCREMENT=12 DEFAULT CHARSET=latin1;
";
		echo "Creating table: gearman_functions\n";
		$this->execute($sql);

$sql = "
CREATE TABLE `gearman_job_servers` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `schema_id` int(11) NOT NULL,
  `hostname` varchar(50) NOT NULL,
  `port` smallint(5) unsigned NOT NULL,
  `last_updated` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `schema_id_hostname_port_unique` (`schema_id`,`hostname`,`port`),
  KEY `schema_id` (`schema_id`),
  CONSTRAINT `gearman_job_servers_ibfk_1` FOREIGN KEY (`schema_id`) REFERENCES `schemata` (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=22 DEFAULT CHARSET=latin1;
";
		echo "Creating table: gearman_job_servers\n";
		$this->execute($sql);

#FIXME: structure will change
$sql = "
CREATE TABLE `gearman_workers` (
  `pid` int(11) NOT NULL,
  `function_id` int(11) DEFAULT NULL,
  `last_updated` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  PRIMARY KEY (`id`),
  UNIQUE KEY `function_id` (`function_id`,`pid`),
  CONSTRAINT `gearman_workers_ibfk_1` FOREIGN KEY (`function_id`) REFERENCES `gearman_functions` (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=latin1;
";

		echo "Creating table: gearman_workers\n";
		$this->execute($sql);

$sql = "
CREATE VIEW `shard_columns` AS select `column_sequences`.`id` AS `id`,`column_sequences`.`id` AS `column_id`,`column_sequences`.`schema_id` AS `schema_id`,`column_sequences`.`sequence_name` AS `column_name`,`column_sequences`.`next_value` AS `next_value`,`column_sequences`.`datatype` AS `datatype` from `column_sequences` where (`column_sequences`.`sequence_type` = 'shard_column') 
WITH CASCADED CHECK OPTION;";
		echo "Creating view: shard_columns\n";
		$this->execute($sql);

	}

	public function get_params($schema_name = null) {

		if(isset($schema_name)) {
			$subquery = "(select id from schemata where schema_name='" . $this->escape($schema_name) . "' limit 1)";
		} else {
			$subquery = "(select id from schemata where is_default_schema = true limit 1)";
		}
	
		$params = array();	
		$sql = "select var_name, var_value from schemata_config where schema_id = $subquery";
		$stmt = $this->execute($sql);
		while($row = $this->conn->my_fetch_assoc($stmt)) {
			$params[$row['var_name']] = $row['var_value'];	
		}
		return $params;

	}
	
	public function get_shards($schema_name = null, $coord_only = false) {
		if(isset($this->shard_cache[isset($schema_name) ? $schema_name : '.'])) return $this->shard_cache[isset($schema_name) ? $schema_name : '.'];
		if(!isset($schema_name)) {
			$sql = "SELECT shards.*,username as user FROM shards join schemata on schema_id = schemata.id where is_default_schema = true";
		} else {
			$sql = "SELECT shards.*,username as user FROM shards join schemata on schema_id = schemata.id where schema_name = '" . $this->escape($schema_name) . "'";

		}

		if($coord_only) $sql .= " AND coord_shard = true";

		$stmt = $this->execute($sql);
	
		$shards = array();
		while($row = $this->conn->my_fetch_assoc($stmt)) {
			$shard = array();
			foreach($row as $key => $val) {
				if(is_numeric($key)) continue;
				$shard[$key] = $val;
			}
				
			$shards[$row['shard_name']] = $shard;
		}
	
		$this->shard_cache[isset($schema_name) ? $schema_name : '.'] = $shards;

		return $shards;
	}
	
	public function add_shard($shard_name, $shard, $coord_shard = true, $accepts_new_rows = true, $extra_info = "NULL", $schema_name = null) {
                if(empty($shard['user'])) $username = "NULL"; else $username="'" . $this->escape($shard['user']) . "'";
                if(empty($shard['db'])) $db = "NULL"; else $db="'" . $this->escape($shard['db']) . "'";
                if(empty($shard['password'])) $password = "NULL"; else $password="'" . $this->escape($shard['password']) . "'";
                if(empty($shard['port'])) $port = "NULL"; else $port="'" . $this>escape($shard['port']) . "'";
                if(empty($shard['host'])) $host = "NULL"; else $host="'" . $this->escape($shard['host']) . "'";
                if(empty($shard['shard_rdbms'])) $shard_rdbms = "'mysql'"; else $shard_rdbms="'" . $shard['shard_rdbms'] . "'";

		if(isset($schema_name)) {
			$subquery = "(select id from schemata where schema_name='" . $this->escape($schema_name) . "' limit 1)";
		} else {
			$subquery = "(select id from schemata where is_default_schema = true limit 1)";
		}

		$shard_name = $this->escape($shard_name);
		if($extra_info != "NULL" )$extra_info = $this->escape($extra_info);
		$coord_shard = $this->escape($coord_shard);
		

                $sql = "INSERT INTO shards (schema_id, shard_name, extra_info, shard_rdbms, coord_shard, accepts_new_rows, username, password, host, port, db) VALUES(";
                $sql .= "$subquery, '$shard_name',$extra_info,$shard_rdbms,'$coord_shard','$accepts_new_rows','$username',$password,$host,$port,$db)";

                $this->execute($sql);

		$this->refresh_shard_cache($schema_name);
	}

	public function refresh_shard_cache($schema_name = null) {
		$this->shard_cache[isset($schema_name) ? $schema_name : '.' ] = null;
		$this->get_shards($schema_name);
		return true;
	}

	public function remove_shard($shard_name, $schema_name = null) {
		if(isset($schema_name)) {
			$schema_name = $this->escape($schema_name);
			$sql = "delete shards.* from shards join schemata on schema_id = schemata.id where schema_name = '$schema_name' and shard_name = '$shard_name'";
		} else {
			$sql = "delete shards.* from shards join schemata on schema_id = schemata.id where is_default_schema = true and shard_name = '$shard_name'";
		}
		$this->execute($sql);
		$this->refresh_shard_cache($schema_name);
		return true;
	}
	
	public function drop_store() {
		$sql = "DROP DATABASE IF EXISTS " . $this->config['db'];
		$this->execute($sql);
		return true;
	}

	public function add_schema($schema_name, $is_default_schema = 1) {

		#this will throw an exception if the unique key is violated	
		$sql = "insert into schemata (schema_name, is_default_schema) values('" . $this->escape($schema_name) . "', $is_default_schema)";
		$this->execute($sql);

		return true;
	
	}

	public function add_column($column, $schema_name = null,$datatype = 'integer',$next_value=1) {

		if(isset($schema_name)) {
			$subquery = "(select id from schemata where schema_name='" . $this->escape($schema_name) . "' limit 1)";
		} else {
			$subquery = "(select id from schemata where is_default_schema = true limit 1)";
		}

		$sql =  "INSERT INTO column_sequences (schema_id, sequence_name, next_value, datatype) VALUES (";
		$sql .= "$subquery,'$column',1,'$next_value','$datatype')";

		$this->execute($sql);

		return true;

	}

	public function move_key($column, $key, $tables, $dest_shard, $schema_name = null ) {
		throw new Exception('Not yet implemented.');
	}

	public function lock_key($column, $key, $lock_mode, $schema_name = null) {
		throw new Exception('Not implemented.');
	}

	public function unlock_key($column, $key, $schema_name = null) {
		throw new Exception('Not implemented.');
	}

	public function get_column_info($column_name, $schema_name = null) {

		$column = $this->escape($column);

		if(isset($schema_name)) {
			$subquery = "(select id from schemata where schema_name='" . $this->escape($schema_name) . "' limit 1)";
		} else {
			$subquery = "(select id from schemata where is_default_schema = true limit 1)";
		}

		$sql = "select * from column_sequences where schema_id = $subquery and sequence_name='$column' and sequence_type = 'shard_column' FOR UPDATE";
		$stmt = $this->execute($sql);

		$row = $this->conn->my_fetch_assoc($stmt);
		if(!$row) throw new Exception('Specified column is not a shard key in the given schema');

		return $row;

	}


	public function new_key($column, $key = null, $shard_name = null, $schema_name = null, $col_info = null) {

		if(!isset($col_info)) $col_info = $this->get_column_info($column, $schema_name);		

		if(isset($key)) {

			if($col_info['datatype'] != 'string') {
				if($col_info['next_value'] < $key) {
					$sql = "INSERT INTO column_sequences (schema_id, sequence_name,next_value) values (";
					$sql .= "$subquery, '$column', 1) ON DUPLICATE KEY UPDATE next_value = last_insert_id($key)";
				} 
			}

		} else {

			#increment non-string columns.  this query won't do anything on string columns (0 rows returned by select for string columns)
			$sql = "INSERT INTO column_sequences (schema_id, sequence_name,next_value) VALUES(";
			$sql .= "$subquery, '$column', 1) ";
			$sql .= "ON DUPLICATE KEY UPDATE next_value = last_insert_id(next_value+1)";
			$this->execute($sql);
			$key = $this->my_insert_id();

		}

		
		if(!isset($shard_name)) {
			$sql = "select id, shard_name from shards where schema_id = $subquery order by rand() limit 1";
		} else {
			$sql = "select id, shard_name from shards where schema_id = $subquery and shard_name = '$shard_name' limit 1";
		}
		$stmt = $this->execute($sql);
		$row = $this->conn->my_fetch_assoc($stmt);
		if(!$row) throw new Exception('Could not locate the shard for the row');

		$shard_id = $row['id'];
		$shard_name = $row['shard_name'];

		if($row['datatype'] != 'string') {
			$sql = "INSERT INTO shard_map (column_id, shard_id, key_value) VALUES (";
			$sql .= "$shard_id_sq, $subquery, $key);";
		} else {
			$sql = "INSERT INTO shard_map (column_id, shard_id, key_string_value) VALUES (";
			$sql .= "$shard_id_sq, $subquery, $key);";
		}

		$this->execute($sql);
		        
		return array($shard_name => $id);


	}

	#returns an array which maps shard names to id values
	#will create the mapping if it does not exist

	/*
	output:
	$map[] = array('shard1' => 1), array('shard20', 5), ...

	You can map with LIKE, IN, equals(=) or null safe equals(<=>).  
	$key should be an IN list "(1,2,3,4,5)" if the operator is IN

	*/
	public function &map($column, $key, $schema_name=null, $operator = "=", $col_info  = null, $create_on_miss = true) {

		if(!isset($col_info)) $col_info = $this->get_column_info($column, $schema_name);		

		if($col_info['datatype'] != 'string') {
			if($operator == "=" || $operator == "<=>") {
				if(!is_numeric($key)) throw new Exception('Tried to map a string value to an integer mapping');
			} 
			$sql = "select shard_name,key_value as val from shard_map join shards on shard_id = shards.id where column_id = {$col_info['id']} and key_value {$operator} {$key}";
		} else {
			$sql = "select shard_name,key_string_value as val from shard_map join shards on shard_id = shards.id where column_id = {$col_info['id']} and key_string_value {$operator} {$key}";
		}

		$stmt = $this->execute($sql);	

		$map = array();

		while($row = $this->my_fetch_assoc($stmt)) {
			if(empty($map[$row['shard_name']])) $map[$row['shard_name']] = array();
			$map[$row['shard_name']] = $row['val'];
		}
		

		if(empty($map)) {

			if($create_on_miss) {
				#map to a random shard
				$info = $this->new_key($column, $key, null, $schema_name, $col_info);
				return $info;
			} 

			return false;
			
		}

		return($map);
	
	}

	#get all the keys which belong to a particular shard
	public function &get_keys($column, $shard_name, $schema_name = null, $col_info = null) {
		
		if(!isset($col_info)) $col_info = $this->get_column_info($column, $schema_name);

		if($col_info['datatype'] != 'string') {
			$sql = "select key_value val from shard_map join shards on shards_id = shards.id where shard_name = '$shard_name' and column_id = {$col_info['id']}";
		} else {
			$sql = "select key_string_value val from shard_map join shards on shards_id = shards.id where shard_name = '$shard_name' and column_id = {$col_info['id']}";
		}
		$stmt = $this->execute($sql);	
		$keys = array();

		while($row = $this->my_fetch_assoc($stmt)) {
			$keys[] = $row['val'];
		}
	
		return $keys;
	}

	public function &get_key_distribution($column, $schema_name = null) {
		
		if(isset($schema_name)) {	
			$schema_name = $this->escape($schema_name);
			$sql = "select shard_name, count(*) cnt from shard_map join shards on shard_id = shards.id join column_sequences on column_id = column_sequences.id where column_name = '$column' and sequence_type = 'shard_column' and schema_name = '$schema_name' GROUP BY shard_name";
		} else {
			$sql = "select shard_name, count(*) cnt from shard_map join shards on shard_id = shards.id join column_sequences on column_id = column_sequences.id where column_name = '$column' and sequence_type = 'shard_column' and is_default_schema = true GROUP BY shard_name";

		}

		$stmt = $this->execute($sql);
		while($row = $this->my_fetch_assoc($stmt)) {
			$rows[$row['shard_name']] = $row['cnt'];
		}
	
		return $rows;
	}


	public function get_column($schema_name = null) {
		if(!isset($schema_name)) {
			$sql = "select column_name from shard_columns join schemata on schema_id = schemata.id where is_default_schema = true";
		} else {
			$sql = "select column_name from shard_columns join schemata on schema_id = schemata.id where schema_name = '$schema_name'";
		}
		$stmt = $this->execute($sql);	
		while( $row = $this->conn->my_fetch_assoc($stmt) ) {
			$column = $row['column_name'];
		}
		return $column;
	}

	public function get_gearman_servers($schema_name = null) {
		if(isset($schema_name)) {
			$subquery = "(select id from schemata where schema_name='" . $this->escape($schema_name) . "' limit 1)";
		} else {
			$subquery = "(select id from schemata where is_default_schema = true limit 1)";
		}

		$sql = "select concat(hostname, ':', port) as server from gearman_job_servers where schema_id = $subquery and enabled = 1";
		$stmt=$this->execute($sql);
		$servers = array();
		while( $row = $this->conn->my_fetch_assoc($stmt) ) {
			$servers[] = $row['server'];
		}
		return $servers;

	}
	
}

class HashShardKeyMapper extends DirectoryShardKeyMapper{

	public function __construct($config, $no_connect = false ) {
		parent::__construct($config);
	} 

	public function remove_shard($shard_id, $schema = null) {
		throw new Exception('Can not remove shard: Removing a shard would change the hash distribution');
	}
	
	public function move_key($column, $key, $tables, $dest_shard, $schema_name = null ) {
		throw new Exception('Hash partitions are fixed.');
	}

	public function lock_key($column, $key, $lock_mode, $schema = null) {
		throw new Exception('Locks are not supported.');
	}

	public function unlock_key($column, $key, $schema = null) {
		throw new Exception('Locks are not supported.');
	}

	#get a new key value for the given partition
	public function new_key($column, $key = null, $shard_name = null, $schema_name = null, $col_info = null) {
		throw new Exception('Hash partitioning does not support key generation');
	}

	#get all the keys which belong to a particular shard
	public function &get_keys($column, $shard_name, $schema_name = null, $col_info = null) {
		throw new Exception('Hash partitioning does not use a data store and can not enumerate keys');
	}

	public function &get_key_distribution($column, $schema_name = null) {
		throw new Exception('Hash partitioning does not use a data store and can not compute the distribution of keys');
	}

	#returns a shard object
        public function &map($column, $key, $schema_name=null, $operator = "=", $col_info  = null, $create_on_miss = true) {
		static $cache = array();
		$shards = $this->get_shards($schema_name);

		if($operator != '=' && $operator != 'IN') throw new Exception("$operator is not supported in this mapper");

		if($operator == "IN") {
			$vals = explode(',', trim($key,'()'));
			$prev_item = "";
			$list = array();
			foreach($list as $item) {
				if($prev_item != "") $item = $prev_item . $item;
				#did explode split a string containing an embedded comma?
				if(!is_numeric($item) && !in_array(substr(trim($item), -1), array("'",'"'))) {
					$prev_item = $item;
				} else {
					$prev_item = "";
				}

				$list[] = $item;
			}

		} else {
			$list = array($key);
		}

		$map = array();
		$shard_names = array_keys($shards);
		foreach($list as $item) {

			if(is_numeric($item)) {
				$map[] = array($shard_names[ $item % count($shard_names) ] => $item);
			} else {
				$map[] = array($shard_names[ (abs(crc32($item)) % count($shard_names)) ] => $item);
			}

		}

		return $map;
	}



}
